package simpleflink;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class WindowTest1_TimeWindow {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//
//        env.setParallelism(1);
//
//        DataStream<String> inputStream = env.readTextFile("data/sensor.txt");
//
//        DataStream<SensorReading> dataStream = inputStream.map(line -> {
//            String[] fields = line.split(",");
//            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
//        });
//
//        //1.增量聚合函数(这里简单统计每隔key组里传感器信息总数)
//        DataStream<Integer> resultStream = dataStream.keyBy("id")
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
//                //输入值、累加器(中间值)、输出值
//                .aggregate(new AggregateFunction<SensorReading, Integer, Integer>() {
//
//                    // 新建的累加器
//                    @Override
//                    public Integer createAccumulator() {
//                        return 0;
//                    }
//
//                    // 每个数据在上次的基础上累加
//                    @Override
//                    public Integer add(SensorReading value, Integer accumulator) {
//                        return accumulator + 1;
//                    }
//
//                    // 返回结果值
//                    @Override
//                    public Integer getResult(Integer accumulator) {
//                        return accumulator;
//                    }
//
//                    // 分区合并结果(TimeWindow一般用不到，SessionWindow可能需要考虑合并)
//                    @Override
//                    public Integer merge(Integer a, Integer b) {
//                        return a + b;
//                    }
//                });
//        //DataStream<String> inputStream = env.socketTextStream("node01",9999);
//        resultStream.print("result");
//        env.execute();
    }
}
